package com.navinfo.platform.demo.service.impl;

import com.mongodb.spark.rdd.api.java.JavaMongoRDD;
import com.navinfo.platform.common.service.ICommonService;
import com.navinfo.platform.common.service.ILoadDataChannel;
import com.navinfo.platform.common.service.ILoadDataService;
import com.navinfo.platform.common.service.impl.LoadDataFromHdfs;
import com.navinfo.platform.common.service.impl.LoadDataFromMongo;
import com.navinfo.platform.common.service.impl.LoadDataFromMysql;
import com.navinfo.platform.common.utils.DateUtils;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.bson.Document;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/***
 * @author gx
 */
public enum DemoLoadDataService implements ILoadDataService {
    INSTANCE;
    private ILoadDataChannel hdfsChannel = LoadDataFromHdfs.INSTANCE;
    private ILoadDataChannel mongoChannel = LoadDataFromMongo.INSTANCE;
    private ILoadDataChannel mysqlChannel = LoadDataFromMysql.INSTANCE;
    @Override
    public List<Object> loadData(SparkSession spark, Map<String, String> configMap) {

        Map<String, String> mysqlConfig = new HashMap<>();
        mysqlConfig.put("url", configMap.get("mysql.url"));
        mysqlConfig.put("tablename", configMap.get("vehicle.sql"));
        mysqlConfig.put("user", configMap.get("mysql.username"));
        mysqlConfig.put("password", configMap.get("mysql.password"));

        Dataset<Row> mysqlData = mysqlChannel.readData(spark, mysqlConfig);

        //ready hdfs param
        String path = configMap.get("hdfs.trip.path");
        Dataset<Row> hdfsData = hdfsChannel.readData(spark, path);

        //ready mongo param
        String mongoCollectionName = configMap.get("mongo.collection.name");
        String condition = configMap.get("mongo.contidion");
        String projection = configMap.get("mongo.projection");
        String day = configMap.get(ICommonService.DAY_VALUE);//defalut yyyyMMdd
        String date = DateUtils.format(day, "yyyyMMdd","yyyy-MM-dd");
        Map<String, String> mongoConfig = new HashMap<>();
        mongoConfig.put("mongoCollectionName",mongoCollectionName);
        mongoConfig.put("collectionSuffix", "_" + day.substring(0,6));
        if(condition != null){
            mongoConfig.put("condition", String.format(condition, date));
        }
        if(projection != null){
            mongoConfig.put("projection", projection);
        }
        JavaMongoRDD<Document> javaMongoRDD = mongoChannel.readData(spark, mongoConfig);
        return Arrays.asList(hdfsData, javaMongoRDD, mysqlData);
    }
}
